package ex.datastream.windows.windowFunctions;

import ex.datastream.ApiFrame;
import ex.datastream.functions.function.MyAggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * keyBy+window+aggregate(AggregateFunction)
 * 输入
 a,1
 a,2
 a,3
 b,1
 c,1
 c,2
 c,3

 * <p>
 * 输出
 2> (a,6)
 1> (b,1)
 1> (c,6)
 */
public class AggregateFuncTest extends ApiFrame {
    public static void main(String[] args) throws Exception {
        AggregateFuncTest windows = new AggregateFuncTest();
        windows.getEnv();
        DataStream<Tuple2<String, Long>> input = windows.getsocketTextStreamText();
        input.keyBy(k -> k.f0)
        .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
        .aggregate(new MyAggregateFunction())
        .print();

        windows.env.execute();
    }
}
